#pragma once

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../common/message.pb.h"
#include "../common/protocol.pb.h"
#include "../common/Log.hpp"
#include "../common/ThreadPool.hpp"
#include "ConnectionManager.hpp"
#include "ConsumerManager.hpp"
#include "VirtualHost.hpp"
#include "MicroMQController.hpp"
#include <iostream>
#include <memory>
#include <functional>
#include <string>
#include <unordered_map>

namespace micromq
{
    class MicroMQServer
    {
    public:
        using ptr = std::unique_ptr<MicroMQServer>;
    public:
        MicroMQServer(uint16_t serverPort, const std::string& baseDir, const std::string& dbFileName, uint16_t controllerPort, std::string resourceDir)
            : _server(&_baseLoop, muduo::net::InetAddress("0.0.0.0", serverPort), "MicroMQServer", muduo::net::TcpServer::kReusePort)
            , _dispatcher(std::bind(&MicroMQServer::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
            , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
            , _host(std::make_shared<VirtualHost>("VirtualHost", baseDir, dbFileName))
            , _consumerManager(std::make_shared<ConsumerManager>())
            , _connectionManager(std::make_shared<ConnectionManager>())
            , _controller(std::make_shared<MicroMQController>(controllerPort, resourceDir, _host))
        {
            _controller->start();
            _server.setConnectionCallback(std::bind(&MicroMQServer::onConnection, this,std::placeholders::_1));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<OpenChannelRequest      >(std::bind(&MicroMQServer::onOpenChannel,      this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<CloseChannelRequest     >(std::bind(&MicroMQServer::onCloseChannel,     this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<DeclareExchangeRequest  >(std::bind(&MicroMQServer::onDeclareExchange,  this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<RemoveExchangeRequest   >(std::bind(&MicroMQServer::onRemoveExchange,   this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<DeclareMsgQueueRequest  >(std::bind(&MicroMQServer::onDeclareMsgQueue,  this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<RemoveMsgQueueRequest   >(std::bind(&MicroMQServer::onRemoveMsgQueue,   this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<BindRequest             >(std::bind(&MicroMQServer::onBind,             this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<UnbindRequest           >(std::bind(&MicroMQServer::onUnbind,           this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<PublishMessageRequest   >(std::bind(&MicroMQServer::onPublishMessage,   this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<AckMessageRequest       >(std::bind(&MicroMQServer::onAckMessage,       this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<FollowMsgQueueRequest   >(std::bind(&MicroMQServer::onFollowMsgQueue,   this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<UnfollowMsgQueueRequest >(std::bind(&MicroMQServer::onUnfollowMsgQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            std::unordered_map<std::string, MsgQueue::ptr> msgQueueMap = _host->getAllMsgQueue();
            for(auto& it : msgQueueMap)
            {
                _consumerManager->createConsumerCenter(it.first); //根据历史数据初始化消费者
                logInfo("history data recovered...");
            }
        }

        void start() 
        {
            logInfo("MicroMQ server start...");
            _server.start();
            _baseLoop.loop();
        }

        ~MicroMQServer()
        {}
    private:
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected()) { 
                logInfo("server connection established...");
                _connectionManager->addConnection(_host, _consumerManager, conn, _codec);
            } else {
                logInfo("server connection closed...");
                _connectionManager->removeConnection(conn);
            }
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, 
            const std::shared_ptr<google::protobuf::Message>& message, muduo::Timestamp)
        {
            logInfo("unknown message: %s, connection shutdown...", message->GetTypeName().c_str());
            conn->shutdown();
        }

        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const OpenChannelRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            connection->openChannel(req);
            logInfo("open channel, id: %s", req->channel_id().c_str());
        }

        void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const CloseChannelRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            connection->closeChannel(req);
            logInfo("close channel, id: %s", req->channel_id().c_str());
        }

        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const DeclareExchangeRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel: %s can not found...", req->channel_id().c_str());
                return;
            }
            channel->declareExchange(req);
            logInfo("declare exchange: %s", req->exchange_name().c_str());
        }

        void onRemoveExchange(const muduo::net::TcpConnectionPtr& conn, const RemoveExchangeRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->removeExchange(req);
            logInfo("remove exchange: %s", req->exchange_name().c_str());
        }

        void onDeclareMsgQueue(const muduo::net::TcpConnectionPtr& conn, const DeclareMsgQueueRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->declareMsgQueue(req);
            logInfo("declare message queue: %s", req->msg_queue_name().c_str());
        }

        void onRemoveMsgQueue(const muduo::net::TcpConnectionPtr& conn, const RemoveMsgQueueRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->removeMsgQueue(req);
            logInfo("remove message queue: %s", req->msg_queue_name().c_str());
        }

        void onBind(const muduo::net::TcpConnectionPtr& conn, const BindRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->bind(req);
            logInfo("bind exchange '%s' with message queue'%s', binding key: %s", 
                req->exchange_name().c_str(), req->msg_queue_name().c_str(), req->binding_key().c_str());
        }

        void onUnbind(const muduo::net::TcpConnectionPtr& conn, const UnbindRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->unbind(req);
            logInfo("unbind exchange '%s' with message queue'%s'", req->exchange_name().c_str(), req->msg_queue_name().c_str());
        }

        void onPublishMessage(const muduo::net::TcpConnectionPtr& conn, const PublishMessageRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->publishMessage(req);
            logInfo("publish message: %s", req->content().c_str());
        }

        void onAckMessage(const muduo::net::TcpConnectionPtr& conn, const AckMessageRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->ackMessage(req);
            logInfo("ack message, id: %s", req->message_id().c_str());
        }

        void onFollowMsgQueue(const muduo::net::TcpConnectionPtr& conn, const FollowMsgQueueRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->followMsgQueue(req);
        }

        void onUnfollowMsgQueue(const muduo::net::TcpConnectionPtr& conn, const UnfollowMsgQueueRequestPtr& req, muduo::Timestamp)
        {
            Connection::ptr connection = _connectionManager->getConnection(conn);
            if(connection.get() == nullptr)
            {
                logError("connection can not found...");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannelById(req->channel_id());
            if(channel.get() == nullptr)
            {
                logError("channel can not found...");
                return;
            }
            channel->unfollowMsgQueue(req);
            logInfo("consumer: %s unfollowed message queue: %s", req->consumer_tag(), req->msg_queue_name());
        }
    private:
        muduo::net::EventLoop _baseLoop;
        muduo::net::TcpServer _server; 
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        
        VirtualHost::ptr _host;
        ConsumerManager::ptr _consumerManager;
        ConnectionManager::ptr _connectionManager;
        MicroMQController::ptr _controller;
    };
}